package com.hudson.broker.handler;

import com.hudson.broker.handler.process.*;
import com.hudson.broker.monitor.ThreadPoolHelper;
import com.hudson.broker.utils.NettyUtil;
import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.enums.MessageTypeEnum;
import io.netty.channel.ChannelHandlerContext;

import java.util.concurrent.ConcurrentHashMap;

public class HandlerProcesserManage {

    /**
     * 存放不同业务对应的处理器,使用线程池异步处理,与netty线程分离,提高netty线程的利用率
     */
    private static final ConcurrentHashMap<String, Process> PROCESSER_CONCURRENT_HASH_MAP = new ConcurrentHashMap<>();


    /**
     * 添加处理器到map中,后续统一重写
     */
    public static void start() {
        PROCESSER_CONCURRENT_HASH_MAP.put(MessageTypeEnum.ADD_TOPIC_QUEUE.getType(), new TopicQueueProcesser());
        PROCESSER_CONCURRENT_HASH_MAP.put(MessageTypeEnum.PUSH_MESSAGE_REQ.getType(), new ConsumerProcess());
        PROCESSER_CONCURRENT_HASH_MAP.put(MessageTypeEnum.SEND_MESSAGE_REQ.getType(), new DefaultProduceProcesser());
        PROCESSER_CONCURRENT_HASH_MAP.put(MessageTypeEnum.QUEUE_MAP.getType(), new QueueMapProcesser());
        PROCESSER_CONCURRENT_HASH_MAP.put(MessageTypeEnum.SEND_RELAY_MESSAGE_REQ.getType(), new RelayProduceProcesser());
        PROCESSER_CONCURRENT_HASH_MAP.put(MessageTypeEnum.CONSUMER_RETRY_MESSAGE.getType(), new ConsumerRetryProcess());
    }


    /**
     *
     * @param type 业务类型
     * @param context netty上下文,负责写会响应
     * @param request 请求消息
     */
    public static void execute(String type, ChannelHandlerContext context, RemotingCommand request) {
        Process process = PROCESSER_CONCURRENT_HASH_MAP.get(type);
        if (process != null) {
            ThreadPoolHelper.DEFAULT_THREAD_POOL.exec(() -> {
                RemotingCommand response = process.process(context, request);
                if(response != null){
                    NettyUtil.write(context.channel(),response);
                }
            });
        }
    }

}
